fetch_fundamental_data.py

Fetches quarterly financial results and valuation ratios for all stocks

Overview

The fetch_fundamental_data.py script retrieves detailed fundamental data including quarterly financial results, annual metrics, and valuation ratios for all stocks listed in master_isin_map.json. This is the second script in the EDL Pipeline.

Purpose

Fetches fundamental financial data including:

• Quarterly results (Revenue, Net Profit, EPS, OPM)
• Historical quarterly comparisons (QoQ, YoY growth)
• Annual metrics (Sales Growth, EPS history)
• Ownership data (Promoter, FII, DII holdings)
• Valuation ratios (P/E, P/B, ROE, ROCE, D/E)

API Endpoint

URL · string · required

https://open-web-scanx.dhan.co/scanx/fundamental

Method · string · required

POST

Timeout · number · default: "30"

Request timeout in seconds

Request Payload

{"data": {"isin": "<ISIN>"}}

Parameters

data.isin · string · required

ISIN code of the security to fetch fundamental data for

The script batches ISINs into groups of 100 and sends them as an array:

{
  "data": {
    "isins": ["INE002A01018", "INE009A01021", ...]
  }
}

Output Files

fundamental_data.json · array

Consolidated fundamental data for all stocks (~35 MB). Each object contains:

• isin - ISIN code
• Symbol - Stock symbol (enriched from master map)
• Name - Company name (enriched from master map)
• quarterly_results - Array of quarterly financial data
• annual_results - Annual financial metrics
• ratios - Valuation and financial ratios
• shareholding - Promoter, FII, DII holding percentages

Function Signature

def fetch_fundamental_data():
    """
    Fetches fundamental data for all ISINs in master_isin_map.json.
    
    Process:
        1. Load ISINs from master_isin_map.json
        2. Batch ISINs into groups of 100
        3. Fetch data for each batch with 0.5s delay
        4. Enrich with Symbol and Name from master map
        5. Save consolidated results
        
    Returns:
        None (writes to fundamental_data.json)
    """

Dependencies

Python Packages · list

• requests - HTTP client
• json - JSON processing
• time - Rate limiting delays

Local Modules · list

• pipeline_utils.get_headers() - Returns standard API headers

Input Files · list

• master_isin_map.json - Generated by fetch_dhan_data.py

Code Example

import requests
import json
import time
from pipeline_utils import get_headers

def fetch_fundamental_data():
    master_map_file = "master_isin_map.json"
    api_url = "https://open-web-scanx.dhan.co/scanx/fundamental"
    output_file = "fundamental_data.json"
    
    headers = get_headers()

    # Load ISINs from Master Map
    with open(master_map_file, "r") as f:
        master_map = json.load(f)

    # Create lookup for enrichment
    isin_lookup = {
        item['ISIN']: {"Symbol": item.get("Symbol"), "Name": item.get("Name")} 
        for item in master_map if item.get('ISIN')
    }
    
    all_isins = list(isin_lookup.keys())
    total_isins = len(all_isins)
    print(f"Loaded {total_isins} ISINs from master map.")

    # Batch processing
    batch_size = 100
    all_fundamental_data = []
    
    for i in range(0, total_isins, batch_size):
        batch_isins = all_isins[i:i + batch_size]
        print(f"Fetching batch {i//batch_size + 1}: {len(batch_isins)} ISINs...")
        
        payload = {"data": {"isins": batch_isins}}
        
        try:
            response = requests.post(api_url, json=payload, headers=headers, timeout=30)
            
            if response.status_code == 200:
                data = response.json()
                
                if data.get('status') == 'success':
                    batch_results = data.get('data', [])
                    if batch_results:
                        # Enrich with Symbol and Name
                        for item in batch_results:
                            item_isin = item.get('isin')
                            if item_isin and item_isin in isin_lookup:
                                item['Symbol'] = isin_lookup[item_isin]['Symbol']
                                item['Name'] = isin_lookup[item_isin]['Name']
                        
                        all_fundamental_data.extend(batch_results)
                        print(f"  Success: Received {len(batch_results)} records.")
        except Exception as e:
            print(f"  Exception fetching batch: {e}")
            
        time.sleep(0.5)  # Rate limiting

    # Save consolidated data
    with open(output_file, "w") as f:
        json.dump(all_fundamental_data, f, indent=4)
    print(f"Saved {len(all_fundamental_data)} securities to {output_file}")

Usage

python3 fetch_fundamental_data.py

Performance

• Execution Time: ~2-3 minutes for 2,775 stocks
• API Calls: ~28 requests (100 ISINs per batch)
• Output Size: ~35 MB
• Concurrency: Sequential with 0.5s delay between batches
• Batch Size: 100 ISINs per request

Notes

• Requires master_isin_map.json to exist (run fetch_dhan_data.py first)
• Data is enriched with Symbol and Name for easier cross-referencing
• 30-second timeout per request to handle API delays
• 0.5-second delay between batches to avoid rate limiting